/*
 * All rights Reserved, Designed By Yealink
 * Copyright (c) 2018/1/18 by Yealink
 *
 * @author yl1633
 * Createdate:  2018/1/18 11:47
 */
package com.xx.chat_room.netty;


import com.fasterxml.jackson.databind.ObjectMapper;
import com.xx.chat_room.vo.ChatMessage;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
import io.netty.util.AttributeKey;
import io.netty.util.concurrent.GlobalEventExecutor;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author fanjj
 * @date 2018/1/18
 */
public class TextWebSocketFrameHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {
    private static Logger logger = LoggerFactory.getLogger(TextWebSocketFrameHandler.class);
    private ObjectMapper mapper = new ObjectMapper();
    private static Map<String, ChannelGroup> channelMap = new ConcurrentHashMap<>();
    public static ChannelGroup channels = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) throws Exception {
        Channel incoming = ctx.channel();
        logger.info("Client:" + incoming.remoteAddress() + "发送" + msg.text());
        String roomId = (String) incoming.attr(AttributeKey.valueOf(incoming.id().asShortText())).get();
        if (StringUtils.isBlank(roomId)) {
            return;
        }
        String name = incoming.attr(AttributeKey.valueOf("ip")).get() + "(" + incoming.attr(AttributeKey.valueOf("count")).get() + ")";
        ChannelGroup channelGroup = channelMap.get(roomId);
        ChatMessage chatMessage = new ChatMessage();
        chatMessage.setName(name);
        chatMessage.setTime(System.currentTimeMillis());
        chatMessage.setNumberOfChatting(channelGroup.size());
        channelMap.get(roomId).writeAndFlush(new TextWebSocketFrame(mapper.writeValueAsString(chatMessage)));
    }

    private static int getMaxCount(ChannelGroup channelGroup, String ip) {
        int count = 0;
        for (Channel channel : channelGroup) {
            int tempCount = (int) channel.attr(AttributeKey.valueOf("count")).get();
            if (ip.equals(channel.attr(AttributeKey.valueOf("ip")).get()) && count < tempCount) {
                count = tempCount;
            }
        }
        return count;
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (evt == WebSocketServerProtocolHandler.ServerHandshakeStateEvent.HANDSHAKE_COMPLETE) {
            ctx.pipeline().remove(HttpRequestHandler.class);
            Channel incoming = ctx.channel();
            String roomId = (String) incoming.attr(AttributeKey.valueOf(incoming.id().asShortText())).get();
            if (StringUtils.isBlank(roomId)) {
                return;
            }
            String ip = incoming.remoteAddress().toString();
            ip = ip.substring(1, ip.indexOf(":"));
            incoming.attr(AttributeKey.valueOf("ip")).set(ip);
            if (channelMap.containsKey(roomId)) {
                ChannelGroup channelGroup = channelMap.get(roomId);
                incoming.attr(AttributeKey.valueOf("count")).set(getMaxCount(channelGroup, ip) + 1);
                channelGroup.add(incoming);
            } else {
                ChannelGroup channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
                incoming.attr(AttributeKey.valueOf("count")).set(1);
                channelGroup.add(incoming);
                channelMap.put(roomId, channelGroup);
            }
            ChatMessage chatMessage = new ChatMessage();
            chatMessage.setNumberOfChatting(channelMap.get(roomId).size());
            channelMap.get(roomId).writeAndFlush(new TextWebSocketFrame(mapper.writeValueAsString(chatMessage)));
            channels.add(incoming);
        } else {
            super.userEventTriggered(ctx, evt);
        }
    }

    @Override
    public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
        Channel incoming = ctx.channel();
        logger.info("Client:" + incoming.remoteAddress() + "加入");
    }

    @Override
    public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
        Channel incoming = ctx.channel();
        for (Channel channel : channels) {
            channel.writeAndFlush(new TextWebSocketFrame("[SERVER] - " + incoming.remoteAddress() + " 离开"));
        }
        channels.remove(ctx.channel());
        logger.info("Client:" + incoming.remoteAddress() + "离开");
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        Channel incoming = ctx.channel();
        logger.info("Client:" + incoming.remoteAddress() + "在线");
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        Channel incoming = ctx.channel();
        logger.info("Client:" + incoming.remoteAddress() + "掉线");
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        Channel incoming = ctx.channel();
        logger.info("Client:" + incoming.remoteAddress() + "异常");
        // 当出现异常就关闭连接
        cause.printStackTrace();
        ctx.close();
    }

}